import asyncio
import os
import subprocess

from apps.global_config import logger


def run_spark_submit():
    # 获取当前文件的绝对路径
    current_file_path = os.path.abspath(__file__)

    # 获取当前文件所在的目录
    current_directory = os.path.dirname(current_file_path)

    # 获取上层目录
    parent_directory = os.path.dirname(current_directory)

    properties_file = os.path.join(parent_directory, "jar", "spark.properties")
    jar_path = os.path.join(parent_directory, "jar", "cclp-etl-sarpk-0.1-SNAPSHOT.jar")
    command = [
        '/usr/local/vipcloud/spark/bin/spark-submit',
        '--class', 'com.cqvip.cclp.core.CclpUpdateMgr',
        '--master', 'local[6]',
        '--executor-memory', '5g',
        '--driver-memory', '5g',
        '--total-executor-cores', '6',
        '--properties-file', properties_file,
        jar_path
    ]
    try:
        # 执行命令
        result = subprocess.run(command, check=True, text=True, capture_output=True)
        logger.info(f"spark stdout: {result.stdout}")
    except subprocess.CalledProcessError as e:
        logger.error(f"spark error: {e}")
        logger.info(f"spark output: {e.output}")


### 以上为同步调用方法


async def run_subprocess():
    # 获取当前文件的绝对路径
    current_file_path = os.path.abspath(__file__)

    # 获取当前文件所在的目录
    current_directory = os.path.dirname(current_file_path)

    # 获取上层目录
    parent_directory = os.path.dirname(current_directory)

    properties_file = os.path.join(parent_directory, "jar", "spark.properties")
    jar_path = os.path.join(parent_directory, "jar", "cclp-etl-sarpk-0.1-SNAPSHOT1.jar")
    # command = [
    #     '/usr/local/vipcloud/spark/bin/spark-submit',
    #     '--class', 'com.cqvip.cclp.core.CclpUpdateMgr',
    #     '--master', 'local[6]',
    #     '--executor-memory', '5g',
    #     '--driver-memory', '5g',
    #     '--total-executor-cores', '6',
    #     '--properties-file', properties_file,
    #     jar_path
    # ]
    command = [
        'conda', 'env', 'list'
    ]

    # 创建子进程
    process = await asyncio.create_subprocess_exec(
        *command,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )

    # 等待子进程完成
    stdout, stderr = await process.communicate()

    # 处理子进程的输出
    print(f"stdout: {stdout.decode()}")
    print(f"stderr: {stderr.decode()}")


if __name__ == "__main__":
    # 创建事件循环
    loop = asyncio.get_event_loop()

    # 创建任务并将其加入事件循环
    task = loop.create_task(run_subprocess())

    try:
        # 在任务完成之前保持事件循环运行
        loop.run_until_complete(task)
    finally:
        # 关闭事件循环
        loop.close()
